package com.rasou.jms;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.geektimes.event.EventListener;
import org.geektimes.event.reactive.stream.ListenerSubscriberAdapter;
import org.geektimes.reactive.streams.SimplePublisher;

import javax.jms.*;
import java.util.EventObject;
import java.util.concurrent.CountDownLatch;

public class JmsReceiveListener implements Runnable{
    private final SimplePublisher<EventObject> publisher = new SimplePublisher<>();

    private final ActiveMQConnectionFactory activeMQConnectionFactory;

    public JmsReceiveListener(String url, EventListener listener) {
        this.activeMQConnectionFactory = new ActiveMQConnectionFactory(url);
        publisher.subscribe(new ListenerSubscriberAdapter(listener));
        Runtime.getRuntime().addShutdownHook(new Thread(()->{
            countDownLatch.countDown();
        }));
    }

    private CountDownLatch countDownLatch = new CountDownLatch( 1 );

    @Override
    public void run() {
        Connection connection = null;
        Session session = null;
        MessageConsumer messageConsumer = null;
        try {
            try {
                connection = activeMQConnectionFactory.createConnection();
                connection.start();

                session = connection.createSession( false, Session.AUTO_ACKNOWLEDGE );

                Topic topic = session.createTopic( Constant.TOPIC_NAME );

                messageConsumer = session.createConsumer( topic );
                messageConsumer.setMessageListener( message -> {
                    if ( null != message && message instanceof TextMessage ) {
                        TextMessage objectMessage = (TextMessage) message;
                        try {
                            publisher.publish( new EventObject( objectMessage.getText() ) );
                        } catch ( JMSException e ) {
                            e.printStackTrace();
                        }
                    }
                } );
                //阻塞，
                countDownLatch.await();
            } finally {
                if ( messageConsumer != null ) {
                    messageConsumer.close();
                }
                if ( session != null ) {
                    session.close();
                }
                if ( connection != null ) {
                    connection.close();
                }
            }
        } catch ( Exception e ) {
            e.printStackTrace();
        }
    }
}
